package com.example.sockjsclient.sockjs;

import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.simp.stomp.*;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.messaging.WebSocketStompClient;

import java.io.IOException;
import java.lang.reflect.Type;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;

@Service
public class SockJSClientService {
    private final WebSocketStompClient stompClient;

    // 添加配置注入
    @Value("${rabbitmq.host}")
    private String rabbitmqHost;

    @Value("${rabbitmq.port}")
    private int rabbitmqPort;

    @Value("${rabbitmq.username}")
    private String rabbitmqUsername;

    @Value("${rabbitmq.password}")
    private String rabbitmqPassword;

    private final static String QUEUE_NAME = "test_queue";

    @Autowired
    public SockJSClientService(WebSocketStompClient stompClient) {
        this.stompClient = stompClient;
    }

    public void connectAndSubscribe(String url, String destination) {
        StompSessionHandler sessionHandler = new MyStompSessionHandler(url, destination, stompClient);
        // 创建 StompHeaders 并添加 apikey
        StompHeaders headers = new StompHeaders();
//        headers.add("apikey", apiKey);
        stompClient.connect(url, sessionHandler);
    }

    private class MyStompSessionHandler extends StompSessionHandlerAdapter {

        private static final int MAX_RECONNECT_ATTEMPTS = 10;
        private static final long RECONNECT_DELAY = 5000; // 5 seconds
        private int reconnectAttempts = 0;
        private final String url;
        private final String destination;
        private final WebSocketStompClient stompClient;
//        private final String apiKey;

        public MyStompSessionHandler(String url, String destination, WebSocketStompClient stompClient) {
            this.url = url;
            this.destination = destination;
            this.stompClient = stompClient;
//            this.apiKey = apiKey;
        }

        // 使用ConcurrentHashMap记录消息类型最后处理时间
        private final Map<String, Long> MSG_CACHE = new ConcurrentHashMap<>();

        @Override
        public void afterConnected(StompSession session, StompHeaders connectedHeaders) {
            System.out.println("------连接成功------");
            // 重置重连次数
            reconnectAttempts = 0;
            // 订阅目标地址
            session.subscribe("/topic/*", new StompFrameHandler() {
                @Override
                public Type getPayloadType(StompHeaders headers) {
                    return String.class;
                }

                @Override
                public void handleFrame(StompHeaders headers, Object payload) {
                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    String formattedDate = sdf.format(new Date());
//                    System.out.println("收到消息" + formattedDate + ":" + payload);

                    JSONObject jsonObject = JSONObject.parseObject(payload.toString());
                    // 处理消息
                    if (jsonObject.containsKey("job")) {
                        // 特殊消息实时保存
                        sendMessagesToRabbitMQ(payload.toString());
                        System.out.println("保存消息" + formattedDate + ":" + payload);
                    } else if (jsonObject.containsKey("temperature")) {
                        String msgType = null;
                        // 遍历所有键
                        for (String key : jsonObject.keySet()) {
                            if ("temperature".equals(key)) {
                                 msgType = key;
                            }
                        }
                        long currentTime = System.currentTimeMillis();
                        // 检查时间窗口
                        if (MSG_CACHE.containsKey(msgType)) {
                            long lastTime = MSG_CACHE.get(msgType);
                            if (currentTime - lastTime < 5 * 1000L) {
                                return; // N秒内同类型消息丢弃
                            }
                        }

                        // 更新缓存并处理消息
                        MSG_CACHE.put(msgType, currentTime);
                        sendMessagesToRabbitMQ(payload.toString());
                        System.out.println("保存消息" + formattedDate + ":" + payload);
                    }
                }
            });

            // 订阅目标地址
            /*session.subscribe("/topic/messages", new StompFrameHandler() {
                @Override
                public Type getPayloadType(StompHeaders headers) {
                    return String.class;
                }

                @Override
                public void handleFrame(StompHeaders headers, Object payload) {
                    System.out.println("收到的消息:" + payload);
                }
            });*/

            // 发送消息
//            session.send("/app/send", "Hello, Server!");
        }

        private void sendMessagesToRabbitMQ(String message) {
            try {
                // 连接到RabbitMQ
                ConnectionFactory factory = new ConnectionFactory();
                factory.setHost(rabbitmqHost);
                factory.setPort(rabbitmqPort);
                factory.setUsername(rabbitmqUsername);
                factory.setPassword(rabbitmqPassword);

                // 声明队列并发送消息
                try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {
                    channel.queueDeclare(QUEUE_NAME, true, false, false, null);
                    //队列和消息均持久化
                    AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                            .deliveryMode(2)
                            .build();
                    // Topic交换机测试示例
                    channel.exchangeDeclare("test_topic", BuiltinExchangeType.TOPIC);
                    channel.queueBind(QUEUE_NAME, "test_topic", "user.*");
                    channel.basicPublish("test_topic", "user.create", props, message.getBytes());
                } catch (IOException | TimeoutException e) {
                    throw new RuntimeException(e);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        @Override
        public void handleException(StompSession session, StompCommand command, StompHeaders headers, byte[] payload, Throwable exception) {
            System.err.println("异常:" + exception.getMessage());
        }

        @Override
        public void handleTransportError(StompSession session, Throwable exception) {
            System.err.println("传输错误:" + exception.getMessage());
            if (reconnectAttempts < MAX_RECONNECT_ATTEMPTS) {
                reconnectAttempts++;
                System.out.println("尝试第 " + reconnectAttempts + " 次重连...");
                try {
                    Thread.sleep(RECONNECT_DELAY);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                stompClient.connect(url, this);
            } else {
                System.err.println("达到最大重连次数，停止重连。");
            }
        }
    }
}
